package org.example.realtime.jtp.dws.log.job;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableEnvironment;

/**
 * @Author Lianzy
 * @Package org.example.realtime.jtp.dws.log.job
 * @Date 2025/5/21 9:02
 */
public class SerchKeyTest {
    public static void main(String[] args) {
        //执行环境
        TableEnvironment tableEnvironment = TableEnv();
        //输入表input 映射到kafka消息队列
        createTable(tableEnvironment);
        //数据处理
        hand(tableEnvironment);
    }

    private static void hand(TableEnvironment tableEnvironment) {
        Table searchLogTable = tableEnvironment.sqlQuery("select\n" +
                "    page['item'] as full_word\n" +
                "    ,row_time\n" +
                "from dwd_traffic_page_log_kafka_source\n" +
                "where page['item'] is not null\n" +
                "    and page['last_page_id']='search'\n" +
                "    and page['item_type']='keyword'");
//        table.execute().print();
        tableEnvironment.createTemporaryView("search_log_table",searchLogTable);


    }

    private static void createTable(TableEnvironment tenv) {
        tenv.executeSql("CREATE TABLE if not exists dwd_traffic_page_log_kafka_source\n" +
                "(\n" +
                "    `common` MAP<STRING, STRING> COMMENT '公共环境信息',\n" +
                "    `page`   MAP<STRING, STRING> COMMENT '页面信息',\n" +
                "    `ts`     BIGINT,\n" +
                "   row_time as  TO_TIMESTAMP_LTZ(ts, 3) ,\n" +
                "    WATERMARK FOR row_time AS row_time - INTERVAL '0' MINUTE\n" +
                ") WITH (\n" +
                "    'connector' = 'kafka',\n" +
                "    'topic' = 'dwd-traffic-page-log',\n" +
                "    'properties.bootstrap.servers' = 'node101:9092,node102:9092,node103:9092',\n" +
                "    'properties.group.id' = 'gid_dws_traffic_search_keyword',\n" +
                "    'scan.startup.mode' = 'earliest-offset',\n" +
                "    'format' = 'json',\n" +
                "    'json.fail-on-missing-field' = 'false',\n" +
                "    'json.ignore-parse-errors' = 'true'\n" +
                ")");
    }

    private static TableEnvironment TableEnv() {
        EnvironmentSettings settings = EnvironmentSettings.newInstance()
                .inStreamingMode()
                .useBlinkPlanner()
                .build();

        TableEnvironment tabenv = TableEnvironment.create(settings);
        Configuration configuration = tabenv.getConfig().getConfiguration();
        configuration.setString("table.local-time-zone","Asia/Shanghai");
        configuration.setString("table.exec.resource.default-parallelism","1");
        configuration.setString("table.exec.state.ttl","5 s");

        return tabenv;
    }


}
